/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.core.construction;



import com.bff.gaia.unified.model.pipeline.v1.RunnerApi.SideInput;

import com.bff.gaia.unified.sdk.coders.Coder;

import com.bff.gaia.unified.sdk.transforms.ViewFn;

import com.bff.gaia.unified.sdk.transforms.windowing.WindowMappingFn;

import com.bff.gaia.unified.sdk.util.WindowedValue;

import com.bff.gaia.unified.sdk.values.PCollection;

import com.bff.gaia.unified.sdk.values.PCollectionView;

import com.bff.gaia.unified.sdk.values.PValue;

import com.bff.gaia.unified.sdk.values.PValueBase;

import com.bff.gaia.unified.sdk.values.TupleTag;

import com.bff.gaia.unified.sdk.values.WindowingStrategy;



import javax.annotation.Nullable;

import java.util.Map;

import java.util.Objects;



/** A {@link PCollectionView} created from the components of a {@link SideInput}. */

public class RunnerPCollectionView<T> extends PValueBase implements PCollectionView<T> {

  private final TupleTag<Iterable<WindowedValue<?>>> tag;

  private final ViewFn<Iterable<WindowedValue<?>>, T> viewFn;

  private final WindowMappingFn<?> windowMappingFn;

  private final @Nullable

  WindowingStrategy<?, ?> windowingStrategy;

  private final @Nullable

  Coder<?> coder;

  private final transient @Nullable

  PCollection<?> pCollection;



  /** Create a new {@link RunnerPCollectionView} from the provided components. */

  public RunnerPCollectionView(

      @Nullable PCollection<?> pCollection,

      TupleTag<Iterable<WindowedValue<?>>> tag,

      ViewFn<Iterable<WindowedValue<?>>, T> viewFn,

      WindowMappingFn<?> windowMappingFn,

      @Nullable WindowingStrategy<?, ?> windowingStrategy,

      @Nullable Coder<?> coder) {

    this.pCollection = pCollection;

    this.tag = tag;

    this.viewFn = viewFn;

    this.windowMappingFn = windowMappingFn;

    this.windowingStrategy = windowingStrategy;

    this.coder = coder;

  }



  @Override

  public PCollection<?> getPCollection() {

    return pCollection;

  }



  @Override

  public TupleTag<Iterable<WindowedValue<?>>> getTagInternal() {

    return tag;

  }



  @Override

  public ViewFn<Iterable<WindowedValue<?>>, T> getViewFn() {

    return viewFn;

  }



  @Override

  public WindowMappingFn<?> getWindowMappingFn() {

    return windowMappingFn;

  }



  @Override

  public WindowingStrategy<?, ?> getWindowingStrategyInternal() {

    return windowingStrategy;

  }



  @Override

  public Coder<?> getCoderInternal() {

    return coder;

  }



  @Override

  public Map<TupleTag<?>, PValue> expand() {

    throw new UnsupportedOperationException(

        String.format("A %s cannot be expanded", RunnerPCollectionView.class.getSimpleName()));

  }



  @Override

  public boolean equals(Object other) {

    if (!(other instanceof PCollectionView)) {

      return false;

    }

    @SuppressWarnings("unchecked")

    PCollectionView<?> otherView = (PCollectionView<?>) other;

    return tag.equals(otherView.getTagInternal());

  }



  @Override

  public int hashCode() {

    return Objects.hash(tag);

  }

}